package club.sanmubird.springbootrabbitmq.rabbitmq.config;


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;


/**
 * @author Sam
 */
@Slf4j
@Configuration
public class RabbitMqBaseConfig {
	@Value("${spring.rabbitmq.host}")
	private String addresses;

	@Value("${spring.rabbitmq.port}")
	private String port;

	@Value("${spring.rabbitmq.username}")
	private String username;

	@Value("${spring.rabbitmq.password}")
	private String password;

	@Value("${spring.rabbitmq.virtual-host}")
	private String virtualHost;

	/**
	 * 默认采用了java的序列化，性能比较低，而且阅读不友好
	 */
	@Bean
	public MessageConverter getMessageConverter() {
		return new Jackson2JsonMessageConverter();
	}


	@Bean(name = "demoConnectionFactory")
	public ConnectionFactory connectionFactory() {
		CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
		cachingConnectionFactory.setAddresses(addresses + ":" + port);
		cachingConnectionFactory.setUsername(username);
		cachingConnectionFactory.setPassword(password);
		cachingConnectionFactory.setVirtualHost(virtualHost);
		// 开启发布确认机制
		cachingConnectionFactory.setPublisherConfirms(true);
		return cachingConnectionFactory;
	}

	/**
	 * 消费者使用
	 */
	@Bean(name = "demoSimpleListenerContainerFactory")
	public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(@Qualifier("demoConnectionFactory") ConnectionFactory connectionFactory) {
		SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory = new SimpleRabbitListenerContainerFactory();
		simpleRabbitListenerContainerFactory.setMessageConverter(getMessageConverter());
		simpleRabbitListenerContainerFactory.setConnectionFactory(connectionFactory);
		// 同时最多 处理数量 类似于 channel.basicQos(1)
		simpleRabbitListenerContainerFactory.setPrefetchCount(1);
		// 最小消费者数量
		simpleRabbitListenerContainerFactory.setConcurrentConsumers(1);
		// 最大消费者数量
		simpleRabbitListenerContainerFactory.setMaxConcurrentConsumers(1);
		// 应答模式
		simpleRabbitListenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		return simpleRabbitListenerContainerFactory;
	}


	/**
	 * 生产者使用
	 */
	@Bean(name = "demoRabbitTemplate")
	@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
	public RabbitTemplate rabbitTemplate(@Qualifier("demoConnectionFactory") ConnectionFactory connectionFactory) {
		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

		// 依据RabbitMQ和Spring AMQP参考文档，事务(Transactional)或发布确认(Publisher Confirms / aka Publisher Acknowledgements)机制可保证消息被正确投递，即从理论上来说MQ不会丢消息。
		// 注意这两种机制不能共存。事务机制是重量级的，同步的，会带来大量开销；发布确认机制则是轻量级的，异步的。
		// 对于发布确认机制，(1) 需置CachingConnectionFactory的publisherConfirms属性为true；(2) 生产者需调用setConfirmCallback(ConfirmCallback callback)，Confirms就会回调给生产者；(3) 消费者需考虑消息去重处理。
		// 这里需要注意的是，一个RabbitTemplate只能支持一个ConfirmCallback。

		// 确认回调
		rabbitTemplate.setConfirmCallback(
				// 当交换机连接不上时,会回调这个方法
				(correlationData, ack, cause)
						-> {
					log.info("correlationData: " + correlationData);
					log.info("ack: " + ack);
					if (!ack) {
						log.error("交换机出现异常....");
					}
				}
		);
		rabbitTemplate.setMandatory(true);
		rabbitTemplate.setReturnCallback(
				// 当路由失败时,会回调这个方法
				(message, replyCode, replyText, exchange, routingKey)
						-> {
					log.info("return callback message：{},code:{},text:{}", message, replyCode, replyText);
					log.error("找不到对应的队列");
				}
		);
		rabbitTemplate.setMessageConverter(getMessageConverter());
		return rabbitTemplate;
	}

}
